package com.whale.cloud.util.hbase;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import lombok.extern.log4j.Log4j2;
import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/** Created by jiaoshiru on 2017/6/1. */
@Log4j2
public class HbaseClient {

  private Connection connection;

  public void setConnection(Connection connection) {
    this.connection = connection;
  }

  /**
   * 创建表
   *
   * @param tableName 表名
   * @param family 列族名
   * @throws Exception
   */
  public void creatTable(String tableName, String[] family) throws Exception {
    Admin admin = connection.getAdmin();
    HTableDescriptor desc = new HTableDescriptor(TableName.valueOf(tableName));

    for (int i = 0; i < family.length; ++i) {
      desc.addFamily(new HColumnDescriptor(family[i]));
    }

    if (admin.tableExists(TableName.valueOf(tableName))) {
      log.debug("Table exists");
    } else {
      admin.createTable(desc);
      log.debug("Create table success");
    }
  }

  /**
   * 新增数据
   *
   * @param rowKey 主键
   * @param tableName 表名
   * @param familyName 列族名
   * @param column 列名
   * @param value 值
   * @return
   */
  public int add(
      String rowKey, String tableName, String familyName, String[] column, String[] value) {
    byte flag = 0;
    Put put = new Put(Bytes.toBytes(rowKey));
    try {
      Table e = connection.getTable(TableName.valueOf(tableName));
      /*
      HColumnDescriptor[] columnFamilies = e.getTableDescriptor().getColumnFamilies();
      for (int i = 0; i < columnFamilies.length; ++i) {
        String _familyName = columnFamilies[i].getNameAsString();
        if (_familyName.equals(familyName)) {*/
      for (int j = 0; j < column.length; ++j) {
        put.addColumn(
            Bytes.toBytes(familyName),
            Bytes.toBytes(column[j].trim()),
            Bytes.toBytes(value[j].trim()));
      }
      //}
      //}
      e.put(put);
      e.close();
      flag = 1;
    } catch (IOException var12) {
      log.debug(var12.getMessage());
      return flag;
    }
    return flag;
  }

  /**
   * 删除数据
   *
   * @param tableName 表名
   * @param rowKey 主键
   * @param falilyName 列族名
   * @param columnName 列名
   * @throws IOException
   */
  public void delete(String tableName, String rowKey, String falilyName, String columnName)
      throws IOException {
    Table table = connection.getTable(TableName.valueOf(tableName));
    Delete deleteColumn = new Delete(Bytes.toBytes(rowKey));
    deleteColumn.deleteColumns(Bytes.toBytes(falilyName), Bytes.toBytes(columnName));
    table.delete(deleteColumn);
    log.debug(falilyName + ":" + columnName + " is deleted");
  }

  public void deleteAllColumn(String tableName, String rowKey) throws IOException {
    Table table = connection.getTable(TableName.valueOf(tableName));
    Delete deleteAll = new Delete(Bytes.toBytes(rowKey));
    table.delete(deleteAll);
    log.info("All columns are deleted");
  }
  /**
   * 更新数据
   *
   * @param tableName 表名
   * @param rowKey 主键
   * @param familyName 列族名
   * @param columnName 列名
   * @param value 值
   * @return
   */
  public int update(
      String tableName, String rowKey, String familyName, String columnName, String value) {
    byte flag = 0;

    try {
      Table e = connection.getTable(TableName.valueOf(tableName));
      Put put = new Put(Bytes.toBytes(rowKey));
      put.addColumn(
          Bytes.toBytes(familyName), Bytes.toBytes(columnName.trim()), Bytes.toBytes(value.trim()));
      e.put(put);
      flag = 1;
      return flag;
    } catch (IOException var8) {
      log.debug(var8.getMessage());
      return flag;
    }
  }

  /**
   * 查询
   *
   * @param tableName 表名
   * @param rowKey 主键
   * @return
   * @throws IOException
   */
  public JSONArray getResultColum(String tableName, String rowKey) throws IOException {
    Get get = new Get(Bytes.toBytes(rowKey));
    Table table = connection.getTable(TableName.valueOf(tableName));
    Result r = table.get(get);
    JSONArray result = new JSONArray();
    KeyValue[] var6 = r.raw();
    int var7 = var6.length;

    for (int var8 = 0; var8 < var7; ++var8) {
      KeyValue keyValue = var6[var8];
      String colum = new String(keyValue.getQualifier());
      String values = new String(keyValue.getValue());
      Long timestamp = new Long(keyValue.getTimestamp());
      values = values.replace("NaN", "0");

      try {
        JSONObject e = JSONObject.parseObject(values);
        Map m = (Map) JSONObject.parseObject(colum, HashMap.class);
        e.putAll(m);
        e.put("timestamp", timestamp);
        result.add(e);
      } catch (Exception var16) {;
      }
    }
    table.close();
    return result;
  }

  /**
   * 查询指定column
   *
   * @param tableName 表名
   * @param rowKey 主键
   * @param columnName 列名
   * @return
   * @throws IOException
   */
  public String getResultColumn(String tableName, String rowKey, String columnName)
      throws IOException {
    Result result = getResultValue(tableName, rowKey);
    String res = "";
    if (result.size() > 0) {
      byte[] f = new byte[] {" ".getBytes()[0]};
      res =
          null
                  == Bytes.toString(
                      null == result.getValue("basic".getBytes(), columnName.getBytes())
                          ? f
                          : result.getValue("basic".getBytes(), columnName.getBytes()))
              ? ""
              : Bytes.toString(result.getValue("basic".getBytes(), columnName.getBytes()));
    }

    if (res == null) {
      res = "";
    }

    res = res.replace("NaN", "0");
    return res.trim();
  }

  /**
   * 查询多个column
   *
   * @param tableName 表名
   * @param rowKey 主键
   * @param columnName 多个列名组合
   * @return
   * @throws IOException
   */
  public JSONArray getResultColumnList(String tableName, String rowKey, List<String> columnName)
      throws IOException {
    JSONArray objects = new JSONArray();
    Result result = getResultValue(tableName, rowKey);
    String res = "";
    if (result.size() > 0) {
      byte[] f = new byte[] {" ".getBytes()[0]};

      for (int i = 0; i < columnName.size(); ++i) {
        res =
            null
                    == Bytes.toString(
                        null
                                == result.getValue(
                                    "basic".getBytes(), ((String) columnName.get(i)).getBytes())
                            ? f
                            : result.getValue(
                                "basic".getBytes(), ((String) columnName.get(i)).getBytes()))
                ? ""
                : Bytes.toString(
                    result.getValue("basic".getBytes(), ((String) columnName.get(i)).getBytes()));
        if (StringUtils.isNotBlank(res)) {
          objects.add(JSONObject.parse(res));
        }
      }
    }

    return objects;
  }

  /**
   * 获取列
   *
   * @param tableName 表名
   * @param rowKey 主键
   * @return
   * @throws IOException
   */
  public Result getResultValue(String tableName, String rowKey) throws IOException {
    Get get = new Get(Bytes.toBytes(rowKey));
    Table table = connection.getTable(TableName.valueOf(tableName));
    Result result = table.get(get);
    table.close();
    return result;
  }

  public Result getResultByColumn(String tableName, String rowKey, String columnName)
      throws IOException {
    Table table = connection.getTable(TableName.valueOf(tableName));
    Get get = new Get(Bytes.toBytes(rowKey));
    get.addColumn(Bytes.toBytes("basic"), Bytes.toBytes(columnName)); // 获取指定列族和列修饰符对应的列
    Result result = table.get(get);
    return result;
  }
}
